-
-
Notifications
You must be signed in to change notification settings - Fork 2.5k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
net: Expose UdpSocket::try_clone #6226
Conversation
Useful for parallel sending.
I know that registering clones of the same socket with the same epoll instance can cause problems under some circumstances, but I do not remember the details. I cannot accept a |
From the manpage:
So far so good.
Each clone gets its own I don't see any other mention of dup'd FDs having notable interactions with epoll. Does that address your concern? |
4b9d15e
to
8f98c3a
Compare
epoll tracks the underlying file description, not the descriptor itself. If you dup the socket and register the new fd with tokio, you just get redundant events on both descriptors. This isn't like SO_REUSEPORT where you basically shard the socket in some ways. You are better off putting it in an There are valid uses of |
Added a test that exercises concurrent reads, just for insurance. In the course of validating the test, I discovered that unlike the |
No, but you could use ReusableBoxFuture from tokio-util? |
That's exactly the goal. I've only got concurrent tasks for outgoing data (so far), so typically operations will complete immediately, and all waiters should be woken when send space becomes newly available.
That makes it quite difficult to use concurrently, because the |
right, i forgot about that limitation |
I'm exploring options to use the multi-waiter-capable |
Managed to solve my problem using an trait UdpPoller {
fn poll_writable(self: Pin<&mut Self>, cx: &mut Context) -> Poll<io::Result<()>>;
}
pin_project! {
struct UdpPollHelper<F, T> {
f: F,
#[pin]
fut: Option<T>,
}
}
impl<F, T> UdpPollHelper<F, T> {
fn new(f: F) -> Self {
Self { f, fut: None }
}
}
impl<F, T> UdpPoller for UdpPollHelper<F, T>
where
F: Fn() -> T,
T: Future<Output = io::Result<()>>,
{
fn poll_writable(self: Pin<&mut Self>, cx: &mut Context) -> Poll<io::Result<()>> {
let mut this = self.project();
if this.fut.is_none() {
this.fut.set(Some((this.f)()));
}
let result = this.fut.as_mut().as_pin_mut().unwrap().poll(cx);
if result.is_ready() {
this.fut.set(None);
}
result
}
} The tokio implementation then creates a poller with: Box::pin(UdpPollHelper::new(move || {
let socket = self.clone();
async move { socket.io.writable().await }
})) This makes the overall runtime abstraction a bit weirder, but it's simple enough to use and implement, so I'm satisfied. Thanks for the guidance! |
Motivation
Applications sending many UDP packets (e.g. QUIC servers) benefit from parallelism, since at least Linux UDP sends scale almost linearly with thread count. However, Tokio's
UdpSocket
only stores one waker per direction, so if backpressure from the kernel send queue suspends multiple senders, only one gets woken.Solution
Cloning the UDP socket should allow concurrent I/O from multiple tasks without wakers getting clobbered.